package org.snail.transport;

import com.google.common.base.Strings;
import com.google.common.collect.Maps;
import org.snail.core.IdFactory;
import org.snail.core.InvokeFuture;
import org.snail.exception.TimeoutException;
import org.snail.serialization.SerializerFactory;
import org.snail.serialization.SerializerType;
import org.snail.transport.protocol.*;

import java.util.Map;

/**
 * Created by fish on 17/4/7.
 */
public class ConnectorDispatcher {

    final static Map<Long, InvokeFuture> futures = Maps.newConcurrentMap();

    /**
     * 正常处理,注意调用此方法,在当前线程中等待。
     *
     * @param connection
     * @param request
     * @param timeout
     */
    public Object invoke(Connection connection, Request request, long timeout){
        Message message = wireRequestMessage(request);
        connection.write(message);
        if (timeout <= 0) {
            return null;
        }
        InvokeFuture future = new InvokeFuture();
        futures.put(message.getHeader().getInvokeId(), future);
        Object value = future.getValue(timeout);
        if (value == null) {
            throw new TimeoutException(String.format("await %s",timeout));
        }
        return value;
    }


    /**
     * 结果集处理,进入多线程处理,防止阻塞client线程。
     *
     * @param message
     */
    public void dispatch(Message message) {
        InvokeFuture invokeFuture = futures.remove(message.getHeader().getInvokeId());
        if (invokeFuture != null) {
            invokeFuture.receive(message);
        }
    }


    /**
     * 在业务线程中构建message对象,提高性能
     *
     * @param request
     */
    private Message wireRequestMessage(Request request) {
        byte[] body = SerializerFactory.getDefaultSerializer().writeObject(request);
        Header header = new Header();
        header.setMessageType(MessageType.REQUEST);
        header.setSerializerType(SerializerType.getDefault().value());
        header.setInvokeId(IdFactory.getInvokeId());
        header.bodyLength(body.length);
        return new Message(header, body);
    }


}
